package beautiful.butterfly.drds.data_exchange.taskgroup;


import beautiful.butterfly.drds.data_exchange.AbstractContainer;
import beautiful.butterfly.drds.data_exchange.constant.Constants;
import beautiful.butterfly.drds.data_exchange.enums.State;
import beautiful.butterfly.drds.data_exchange.exception.CommonErrorCode;
import beautiful.butterfly.drds.data_exchange.exception.DataExchangeException;
import beautiful.butterfly.drds.data_exchange.statistics.JvmInfo;
import beautiful.butterfly.drds.data_exchange.statistics.PerfRecord;
import beautiful.butterfly.drds.data_exchange.statistics.PerfTrace;
import beautiful.butterfly.drds.data_exchange.statistics.Phase;
import beautiful.butterfly.drds.data_exchange.statistics.communication.Communication;
import beautiful.butterfly.drds.data_exchange.statistics.communication.CommunicationTool;
import beautiful.butterfly.drds.data_exchange.statistics.container.communicator.taskgroup.StandaloneTGContainerCommunicator;
import beautiful.butterfly.drds.data_exchange.util.Configuration;
import beautiful.butterfly.drds.data_exchange.util.FrameworkErrorCode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;


public class TaskGroupContainer extends AbstractContainer
{
    private static final Logger LOG = LoggerFactory
            .getLogger(TaskGroupContainer.class);

    /**
     * 当前taskGroup所属jobId
     */
    public long jobId;

    /**
     * 当前taskGroupId
     */
    public int taskGroupId;

    /**
     * 使用的channel类
     */
    public String channelClassName;

    /**
     * task收集器使用的类
     */
    public String taskCollectorClass;

    private TaskMonitor taskMonitor = TaskMonitor.getInstance();

    public TaskGroupContainer(Configuration configuration)
    {
        super(configuration);

        initCommunicator(configuration);

        this.jobId = this.configuration.getLong(
                Constants.container_job_id);
        this.taskGroupId = this.configuration.getInt(
                Constants.container_taskgroup_id);

        this.channelClassName = this.configuration.getString(
                Constants.transport_channel_class);
        this.taskCollectorClass = this.configuration.getString(
                Constants.statistics_collector_plugin_taskclass);
    }

    private void initCommunicator(Configuration configuration)
    {
        super.setContainerCommunicator(new StandaloneTGContainerCommunicator(configuration));

    }

    public long getJobId()
    {
        return jobId;
    }

    public int getTaskGroupId()
    {
        return taskGroupId;
    }

    @Override
    public void start()
    {
        try
        {
            /**
             * 状态check时间间隔，较短，可以把任务及时分发到对应channel中
             */
            int sleepIntervalInMillSec = this.configuration.getInt(
                    Constants.container_taskgroup_sleepinterval, 100);
            /**
             * 状态汇报时间间隔，稍长，避免大量汇报
             */
            long reportIntervalInMillSec = this.configuration.getLong(
                    Constants.container_taskgroup_reportinterval,
                    10000);
            /**
             * 2分钟汇报一次性能统计
             */

            // 获取channel数目
            int channelNumber = this.configuration.getInt(
                    Constants.container_taskgroup_channel);

            int taskMaxRetryTimes = this.configuration.getInt(
                    Constants.container_task_failover_maxretrytimes, 1);

            long taskRetryIntervalInMsec = this.configuration.getLong(
                    Constants.container_task_failover_retryintervalinmsec, 10000);

            long taskMaxWaitInMsec = this.configuration.getLong(Constants.container_task_failover_maxwaitinmsec, 60000);

            List<Configuration> taskConfigs = this.configuration
                    .getListConfiguration(Constants.job_content);

            if (LOG.isDebugEnabled())
            {
                LOG.debug("taskGroup[{}]'s task configs[{}]", this.taskGroupId);
            }

            int taskCountInThisTaskGroup = taskConfigs.size();
            LOG.info(String.format(
                    "taskGroupId=[%d] start [%d] channels for [%d] tasks.",
                    this.taskGroupId, channelNumber, taskCountInThisTaskGroup));

            this.containerCommunicator.registerCommunication(taskConfigs);

            Map<Integer, Configuration> taskConfigMap = buildTaskConfigMap(taskConfigs); //taskId与task配置
            List<Configuration> taskQueue = buildRemainTasks(taskConfigs); //待运行task列表
            Map<Integer, TaskExecutor> taskFailedExecutorMap = new HashMap<Integer, TaskExecutor>(); //taskId与上次失败实例
            List<TaskExecutor> taskExecutorList = new ArrayList<TaskExecutor>(channelNumber); //正在运行task
            Map<Integer, Long> taskStartTimeMap = new HashMap<Integer, Long>(); //任务开始时间

            long lastReportTimeStamp = 0;
            Communication lastTaskGroupContainerCommunication = new Communication();

            while (true)
            {
                //1.判断task状态
                boolean failedOrKilled = false;
                Map<Integer, Communication> communicationMap = containerCommunicator.getCommunicationMap();
                for (Map.Entry<Integer, Communication> entry : communicationMap.entrySet())
                {
                    Integer taskId = entry.getKey();
                    Communication taskCommunication = entry.getValue();
                    if (!taskCommunication.isFinished())
                    {
                        continue;
                    }
                    TaskExecutor taskExecutor = removeTask(taskExecutorList, taskId);

                    //上面从runTasks里移除了，因此对应在monitor里移除
                    taskMonitor.removeTask(taskId);

                    //失败，看task是否支持failover，重试次数未超过最大限制
                    if (taskCommunication.getState() == State.failed)
                    {
                        taskFailedExecutorMap.put(taskId, taskExecutor);
                        if (taskExecutor.supportFailOver() && taskExecutor.getAttemptCount() < taskMaxRetryTimes)
                        {
                            taskExecutor.shutdown(); //关闭老的executor
                            containerCommunicator.resetCommunication(taskId); //将task的状态重置
                            Configuration taskConfig = taskConfigMap.get(taskId);
                            taskQueue.add(taskConfig); //重新加入任务列表
                        } else
                        {
                            failedOrKilled = true;
                            break;
                        }
                    } else if (taskCommunication.getState() == State.killed)
                    {
                        failedOrKilled = true;
                        break;
                    } else if (taskCommunication.getState() == State.succeeded)
                    {
                        Long taskStartTime = taskStartTimeMap.get(taskId);
                        if (taskStartTime != null)
                        {
                            Long usedTime = System.currentTimeMillis() - taskStartTime;
                            LOG.info("taskGroup[{}] taskId[{}] is successed, used[{}]ms",
                                    this.taskGroupId, taskId, usedTime);
                            //usedTime*1000*1000 转换成PerfRecord记录的ns，这里主要是简单登记，进行最长任务的打印。因此增加特定静态方法
                            PerfRecord.addPerfRecord(taskGroupId, taskId, Phase.task_total, taskStartTime, usedTime * 1000L * 1000L);
                            taskStartTimeMap.remove(taskId);
                            taskConfigMap.remove(taskId);
                        }
                    }
                }

                // 2.发现该taskGroup下taskExecutor的总状态失败则汇报错误
                if (failedOrKilled)
                {
                    lastTaskGroupContainerCommunication = reportTaskGroupCommunication(
                            lastTaskGroupContainerCommunication, taskCountInThisTaskGroup);

                    throw DataExchangeException.asDataExchangeException(FrameworkErrorCode.PLUGIN_RUNTIME_ERROR, lastTaskGroupContainerCommunication.getThrowable());
                }

                //3.有任务未执行，且正在运行的任务数小于最大通道限制
                Iterator<Configuration> iterator = taskQueue.iterator();
                while (iterator.hasNext() && taskExecutorList.size() < channelNumber)
                {
                    Configuration taskConfig = iterator.next();
                    Integer taskId = taskConfig.getInt(Constants.task_id);
                    int attemptCount = 1;
                    TaskExecutor lastExecutor = taskFailedExecutorMap.get(taskId);
                    if (lastExecutor != null)
                    {
                        attemptCount = lastExecutor.getAttemptCount() + 1;
                        long now = System.currentTimeMillis();
                        long failedTime = lastExecutor.getTimeStamp();
                        if (now - failedTime < taskRetryIntervalInMsec)
                        {  //未到等待时间，继续留在队列
                            continue;
                        }
                        if (!lastExecutor.isShutdown())
                        { //上次失败的task仍未结束
                            if (now - failedTime > taskMaxWaitInMsec)
                            {
                                markCommunicationFailed(taskId);
                                reportTaskGroupCommunication(lastTaskGroupContainerCommunication, taskCountInThisTaskGroup);
                                throw DataExchangeException.asDataExchangeException(CommonErrorCode.WAIT_TIME_EXCEED, "task failover等待超时");
                            } else
                            {
                                lastExecutor.shutdown(); //再次尝试关闭
                                continue;
                            }
                        } else
                        {
                            LOG.info("taskGroup[{}] taskId[{}] attemptCount[{}] has already shutdown",
                                    this.taskGroupId, taskId, lastExecutor.getAttemptCount());
                        }
                    }
                    Configuration taskConfigForRun = taskMaxRetryTimes > 1 ? taskConfig.clone() : taskConfig;
                    TaskExecutor taskExecutor = new TaskExecutor(this, taskConfigForRun, attemptCount);
                    taskStartTimeMap.put(taskId, System.currentTimeMillis());
                    taskExecutor.doStart();

                    iterator.remove();
                    taskExecutorList.add(taskExecutor);

                    //上面，增加task到runTasks列表，因此在monitor里注册。
                    taskMonitor.registerTask(taskId, this.containerCommunicator.getCommunication(taskId));

                    taskFailedExecutorMap.remove(taskId);
                    LOG.info("taskGroup[{}] taskId[{}] attemptCount[{}] is started",
                            this.taskGroupId, taskId, attemptCount);
                }

                //4.任务列表为空，executor已结束, 搜集状态为success--->成功
                if (taskQueue.isEmpty() && isAllTaskDone(taskExecutorList) && containerCommunicator.collectState() == State.succeeded)
                {
                    // 成功的情况下，也需要汇报一次。否则在任务结束非常快的情况下，采集的信息将会不准确
                    lastTaskGroupContainerCommunication = reportTaskGroupCommunication(
                            lastTaskGroupContainerCommunication, taskCountInThisTaskGroup);

                    LOG.info("taskGroup[{}] completed it's tasks.", this.taskGroupId);
                    break;
                }

                // 5.如果当前时间已经超出汇报时间的interval，那么我们需要马上汇报
                long now = System.currentTimeMillis();
                if (now - lastReportTimeStamp > reportIntervalInMillSec)
                {
                    lastTaskGroupContainerCommunication = reportTaskGroupCommunication(
                            lastTaskGroupContainerCommunication, taskCountInThisTaskGroup);

                    lastReportTimeStamp = now;

                    //taskMonitor对于正在运行的task，每reportIntervalInMillSec进行检查
                    for (TaskExecutor taskExecutor : taskExecutorList)
                    {
                        taskMonitor.report(taskExecutor.getTaskId(), this.containerCommunicator.getCommunication(taskExecutor.getTaskId()));
                    }

                }

                Thread.sleep(sleepIntervalInMillSec);
            }

            //6.最后还要汇报一次
            reportTaskGroupCommunication(lastTaskGroupContainerCommunication, taskCountInThisTaskGroup);


        } catch (Throwable e)
        {
            Communication nowTaskGroupContainerCommunication = this.containerCommunicator.collect();

            if (nowTaskGroupContainerCommunication.getThrowable() == null)
            {
                nowTaskGroupContainerCommunication.setThrowable(e);
            }
            nowTaskGroupContainerCommunication.setState(State.failed);
            this.containerCommunicator.report(nowTaskGroupContainerCommunication);

            throw DataExchangeException.asDataExchangeException(
                    FrameworkErrorCode.RUNTIME_ERROR, e);
        } finally
        {
            if (!PerfTrace.getInstance().isJob())
            {
                //最后打印cpu的平均消耗，GC的统计
                JvmInfo jvmInfo = JvmInfo.getJvmInfo();
                if (jvmInfo != null)
                {
                    jvmInfo.updateJvmInfo(false);
                    LOG.info(jvmInfo.totalString());
                }

                LOG.info(PerfTrace.getInstance().summarizeNoException());
            }
        }
    }

    private Map<Integer, Configuration> buildTaskConfigMap(List<Configuration> data)
    {
        Map<Integer, Configuration> map = new HashMap<Integer, Configuration>();
        for (Configuration taskConfig : data)
        {
            int taskId = taskConfig.getInt(Constants.task_id);
            map.put(taskId, taskConfig);
        }
        return map;
    }

    private List<Configuration> buildRemainTasks(List<Configuration> configurationList)
    {
        List<Configuration> remainTaskList = new LinkedList<Configuration>();
        for (Configuration configuration : configurationList)
        {
            remainTaskList.add(configuration);
        }
        return remainTaskList;
    }

    private TaskExecutor removeTask(List<TaskExecutor> taskExecutorList, int taskId)
    {
        Iterator<TaskExecutor> iterator = taskExecutorList.iterator();
        while (iterator.hasNext())
        {
            TaskExecutor taskExecutor = iterator.next();
            if (taskExecutor.getTaskId() == taskId)
            {
                iterator.remove();
                return taskExecutor;
            }
        }
        return null;
    }

    private boolean isAllTaskDone(List<TaskExecutor> taskExecutorList)
    {
        for (TaskExecutor taskExecutor : taskExecutorList)
        {
            if (!taskExecutor.isTaskFinished())
            {
                return false;
            }
        }
        return true;
    }

    private Communication reportTaskGroupCommunication(Communication lastTaskGroupContainerCommunication, int taskCount)
    {
        Communication nowTaskGroupContainerCommunication = this.containerCommunicator.collect();
        nowTaskGroupContainerCommunication.setTimestamp(System.currentTimeMillis());
        Communication reportCommunication = CommunicationTool.getReportCommunication(nowTaskGroupContainerCommunication,
                lastTaskGroupContainerCommunication, taskCount);
        this.containerCommunicator.report(reportCommunication);
        return reportCommunication;
    }

    private void markCommunicationFailed(Integer taskId)
    {
        Communication communication = containerCommunicator.getCommunication(taskId);
        communication.setState(State.failed);
    }

}
